package com.alinesno.cloud.compoment.kafka.config;

import java.nio.charset.Charset;
import java.util.Map;
import java.util.Properties;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;

import cn.hutool.setting.dialect.Props;

/**
 * kafka 发送配置
 * 
 * @author WeiXiaoJin
 * @since 2019年4月9日 上午11:15:09
 */
public class KafkaProducerConfig extends Config {

	private static final Logger log = LoggerFactory.getLogger(KafkaProducerConfig.class);

	@Bean
	public ProducerFactory<String, String> producerFactory() {
		return new DefaultKafkaProducerFactory<>(producerConfigs());
	}

	@Bean
	public Map<String, Object> producerConfigs() {
		Properties properties = Props.getProp("classpath:kafka-producer.properties", Charset.forName("UTF-8"));

		Map<String, Object> props = toMap(properties); // KafkaProperties.buildProducerProperties() ;

//        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
//        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
		
		log.debug("kafak 配置:{}" , props);

		return this.producerConfigs(props);
	}

	@Bean
	public Map<String, Object> producerConfigs(Map<String, Object> props) {
		return props;
	}

	@Bean
	public KafkaTemplate<String, String> kafkaTemplate() {
		return new KafkaTemplate<String, String>(producerFactory());
	}
}